package drds.global_transaction.api.rpc.rpc_client.resource_manager;

import static drds.global_transaction.common.Constants.resourceIds_split_char;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.pool.impl.GenericKeyedObjectPool.Config;

import drds.common.Log;
import drds.global_transaction.api.api.IResourceManager;
import drds.global_transaction.api.api.Resource;
import drds.global_transaction.api.global_transaction_id_context.GlobalTransactionIdContexts;
import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.message_codec.Message;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.RegisterResourceManagerRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.RegisterResourceManagerResponseMessage;
import drds.global_transaction.api.rpc.config.ClientConfig;
import drds.global_transaction.api.rpc.rpc_client.Key;
import drds.global_transaction.api.rpc.rpc_client.Role;
import drds.global_transaction.api.rpc.rpc_client.RpcClient;
import drds.global_transaction.api.rpc.rpc_client.SendMergedWarpMessageRunnable;
import drds.global_transaction.common.GlobalTransactionIds;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.thread.NamedThreadFactory;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.EventExecutorGroup;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Sharable
public final class ResourceManagerRpcClient extends RpcClient
{

	private static final int max_merge_send_thread = 1;
	private static final long keep_alive_time = Integer.MAX_VALUE;
	private static final int max_queue_size = 20000;
	private static final int schedule_interval_mills = 5;
	private static final String merge_thread_prefix = "rpc_merge_message_send";
	private static volatile ResourceManagerRpcClient instance;
	private final ConcurrentMap<String, Object> channelToLockMap = new ConcurrentHashMap<String, Object>();
	private final ConcurrentMap<String, Key> serverAddressToKeyMap = new ConcurrentHashMap<String, Key>();
	private final ConcurrentMap<String, Channel> serverAddressToChannelMap = new ConcurrentHashMap<>();
	private final AtomicBoolean initialized = new AtomicBoolean(false);
	private final ClientConfig clientConfig;
	//
	@Setter
	@Getter
	private IResourceManager resourceManager;
	@Setter
	@Getter
	private String applicationId;
	@Setter
	@Getter
	private String transactionServiceGroup;
	@Setter
	@Getter
	private String customerKeys;

	//
	private ResourceManagerRpcClient(ClientConfig clientConfig, //
			EventExecutorGroup eventExecutorGroup, //
			ThreadPoolExecutor threadPoolExecutor)
	{//
		super(clientConfig, eventExecutorGroup, threadPoolExecutor);
		this.clientConfig = clientConfig;
	}

	public static ResourceManagerRpcClient getInstance(String applicationId, String transactionServiceGroup)
	{
		ResourceManagerRpcClient resourceManagerRpcClient = getInstance();
		resourceManagerRpcClient.setApplicationId(applicationId);
		resourceManagerRpcClient.setTransactionServiceGroup(transactionServiceGroup);
		return resourceManagerRpcClient;
	}

	public static ResourceManagerRpcClient getInstance()
	{
		if (null == instance)
		{
			synchronized (ResourceManagerRpcClient.class)
			{
				if (null == instance)
				{
					ClientConfig clientConfig = new ClientConfig();
					final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(clientConfig.getClientWorkerThreads(), clientConfig.getClientWorkerThreads(), keep_alive_time, TimeUnit.SECONDS, new LinkedBlockingQueue(max_queue_size), new NamedThreadFactory(clientConfig.getRmDispatchThreadPrefix(), clientConfig.getClientWorkerThreads()), new ThreadPoolExecutor.CallerRunsPolicy());
					instance = new ResourceManagerRpcClient(clientConfig, null, threadPoolExecutor);
				}
			}
		}
		return instance;
	}

	@Override
	protected Role getRole()
	{
		return Role.resource_manager_client;
	}

	//
	@Override
	public void init()
	{
		if (initialized.compareAndSet(false, true))
		{
			super.init();
			scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable()
			{
				@Override
				public void run()
				{
					reConnect();
				}
			}, schedule_interval_mills, schedule_interval_mills, TimeUnit.SECONDS);
			ExecutorService threadPoolExecutor = new ThreadPoolExecutor(max_merge_send_thread, max_merge_send_thread, keep_alive_time, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(getThreadPrefix(merge_thread_prefix), max_merge_send_thread));
			threadPoolExecutor.submit(new SendMergedWarpMessageRunnable(this));
		}
	}

	@Override
	protected Config getConfig()
	{
		Config config = new Config();
		config.maxActive = clientConfig.getMaxPoolActive();
		config.minIdle = clientConfig.getMinPoolIdle();
		config.maxWait = clientConfig.getMaxAcquireConnMills();
		config.testOnBorrow = clientConfig.isPoolTestBorrow();
		config.testOnReturn = clientConfig.isPoolTestReturn();
		config.lifo = clientConfig.isPoolFifo();
		return config;
	}

	//
	private void reConnect()
	{
		String[] serverAddresss = globalTransactionServiceGroupAddressManager.lookup(transactionServiceGroup);
		for (String serverAddress : serverAddresss)
		{
			if (serverAddress != null)
			{
				try
				{
					connect(serverAddress);
				} catch (Exception e)
				{
					log.error(FrameworkErrorCode.NetConnect.errCode, "can not connect to " + serverAddress + " cause:" + e.getMessage(), e);
				}
			}
		}
	}

	@Override
	protected Channel connect(String serverAddress)
	{
		Channel channel = serverAddressToChannelMap.get(serverAddress);
		if (channel != null)
		{
			channel = getExistAliveChannel(channel, serverAddress);
			if (null != channel)
			{
				return channel;
			}
		}
		// 再进行获取
		channelToLockMap.putIfAbsent(serverAddress, new Object());
		Object lock = channelToLockMap.get(serverAddress);
		synchronized (lock)
		{
			return connectWithLock(serverAddress);
		}
	}

	private Channel getExistAliveChannel(Channel channel, String serverAddress)
	{
		if (channel.isActive())
		{
			return channel;
		} else
		{
			int i = 0;
			for (; i < ClientConfig.getMaxCheckAliveRetry(); i++)
			{
				try
				{
					Thread.sleep(ClientConfig.getCheckAliveInternal());
				} catch (InterruptedException exx)
				{
					log.error(exx.getMessage());
				}
				channel = serverAddressToChannelMap.get(serverAddress);
				if (null == channel || channel.isActive())
				{
					// 这样就不是中间状态(null != channel && !channel.isActive())
					return channel;
				}
			}
			if (i == ClientConfig.getMaxCheckAliveRetry())
			{
				log.warn("channel " + channel + " is not active after long wait, close it.");
				releaseChannel(channel, serverAddress);
				return null;
			}
		}
		return null;
	}

	private Channel connectWithLock(String serverAddress)
	{
		Channel channel = serverAddressToChannelMap.get(serverAddress);
		if (channel != null && channel.isActive())
		{
			return channel;
		}
		//
		try
		{
			String resourceIds = customerKeys == null ? getResourceIds(resourceManager) : customerKeys;
			if (log.isInfoEnabled())
			{
				log.info("resource_manager_client will register :" + resourceIds);
			}
			// 更新resourceIds
			RegisterResourceManagerRequestMessage registerResourceManagerRequestMessage = null;
			if (null == serverAddressToKeyMap.get(serverAddress))
			{
				registerResourceManagerRequestMessage = new RegisterResourceManagerRequestMessage(applicationId, transactionServiceGroup);
				registerResourceManagerRequestMessage.setResourceIds(resourceIds);
				serverAddressToKeyMap.putIfAbsent(serverAddress, new Key(serverAddress, getRole(), registerResourceManagerRequestMessage));
			} else
			{
				registerResourceManagerRequestMessage = (RegisterResourceManagerRequestMessage) serverAddressToKeyMap.get(serverAddress).getMessage();
				registerResourceManagerRequestMessage.setResourceIds(resourceIds);
			}
			channel = keyToChannelMap.borrowObject(serverAddressToKeyMap.get(serverAddress));
		} catch (Exception exx)
		{
			log.error(FrameworkErrorCode.RegistRM.errCode, "register resource_manager_client failed.", exx);
			throw new FrameworkException("can not register resource_manager_client,err:" + exx.getMessage());
		}
		return channel;
	}

	//
	@Override
	public Object sendMessageWithResponse(Object message) throws TimeoutException
	{
		return sendMessageWithResponse(message, ClientConfig.getRpcRequestTimeout());
	}

	@Override
	public Object sendMessageWithResponse(Object message, long timeout) throws TimeoutException
	{
		return super.sendAsynchronousRequestWithResponse(//
				GlobalTransactionIds.getAddressAndPort(GlobalTransactionIdContexts.getGlobalTransactionIdWithAddressAndPort()), getRemoteServerChannel(), message, timeout);
	}

	private Channel getRemoteServerChannel()
	{
		String serverAddress = null;
		if (!GlobalTransactionIdContexts.inGlobalTransaction())
		{
			return null;
		}
		serverAddress = GlobalTransactionIds.getAddressAndPort(GlobalTransactionIdContexts.getGlobalTransactionIdWithAddressAndPort());
		return connect(serverAddress);
	}

	@Override
	public Object sendMessageWithResponse(String serverAddress, Object message, long timeout) throws TimeoutException
	{
		return super.sendAsynchronousRequestWithResponse(serverAddress, connect(serverAddress), message, timeout);
	}

	@Override
	public void sendResponse(String serverAddress, long messageId, Object message)
	{
		super.sendResponse(messageId, connect(serverAddress), message);
	}

	//
	@Override
	public void userEventTriggered(ChannelHandlerContext channelHandlerContext, Object event)
	{
		if (event instanceof IdleStateEvent)
		{
			IdleStateEvent idleStateEvent = (IdleStateEvent) event;
			//
			if (idleStateEvent == IdleStateEvent.READER_IDLE_STATE_EVENT)
			{
				if (log.isInfoEnabled())
				{
					log.info("ResourceManagerRpcClient channel" + channelHandlerContext.channel() + " idle.");
				}
				try
				{
					keyToChannelMap.invalidateObject(serverAddressToKeyMap.get(channelHandlerContext.channel().remoteAddress()), channelHandlerContext.channel());
				} catch (Exception exx)
				{
					log.error(exx.getMessage());
				} finally
				{
					releaseChannel(channelHandlerContext.channel(), getRemoteAddress(channelHandlerContext));
				}
			}
			//
			if (idleStateEvent == IdleStateEvent.WRITER_IDLE_STATE_EVENT)
			{
				if (log.isInfoEnabled())
				{
					log.info("ResourceManagerRpcClient channel" + channelHandlerContext.channel() + " idle.");
				}
				try
				{
					sendRequest(channelHandlerContext.channel(), HeartbeatMessage.ping);
				} catch (Throwable throwable)
				{
					log.error("", "send request error", throwable);
				}
			}
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception
	{
		if (threadPoolExecutor.isShutdown())
		{
			return;
		}
		if (log.isInfoEnabled())
		{
			log.info("channel inactive:" + channelHandlerContext.channel());
		}
		releaseChannel(channelHandlerContext.channel(), Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress()));
		super.channelInactive(channelHandlerContext);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) throws Exception
	{
		log.error(FrameworkErrorCode.ExceptionCaught.errCode, Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress()) + "connect exception. " + cause.getMessage(), cause);
		releaseChannel(channelHandlerContext.channel(), getRemoteAddress(channelHandlerContext.channel()));
		if (log.isInfoEnabled())
		{
			log.info("remove exception resource_manager_client channel:" + channelHandlerContext.channel());
		}
		super.exceptionCaught(channelHandlerContext, cause);
	}

	//
	public void registerResource(String resourceGroupId, String resourceId)
	{
		synchronized (serverAddressToChannelMap)
		{
			for (Map.Entry<String, Channel> entry : serverAddressToChannelMap.entrySet())
			{
				String serverAddress = entry.getKey();
				Channel channel = entry.getValue();
				sendRegisterMessage(serverAddress, channel, resourceId);
			}
		}
	}

	//
	@Override
	public void onRegisterMessageSuccess(String serverAddress, Channel channel, Message requestMessage, Object response)
	{
		if (customerKeys == null)
		{
			synchronized (serverAddressToChannelMap)
			{
				serverAddressToChannelMap.put(serverAddress, channel);
			}
			//
			String resourceIds = getResourceIds(resourceManager);
			RegisterResourceManagerRequestMessage registerResourceManagerRequestMessage = (RegisterResourceManagerRequestMessage) requestMessage;
			if (registerResourceManagerRequestMessage.getResourceIds() != null)
			{
				if (!registerResourceManagerRequestMessage.getResourceIds().equals(resourceIds))
				{
					sendRegisterMessage(serverAddress, channel, resourceIds);
				}
			}
		}
	}

	private void sendRegisterMessage(String serverAddress, Channel channel, String resourceIds)
	{
		RegisterResourceManagerRequestMessage registerResourceManagerRequestMessage = new RegisterResourceManagerRequestMessage(applicationId, transactionServiceGroup);
		registerResourceManagerRequestMessage.setResourceIds(resourceIds);
		try
		{
			super.asynchronousSendRequestWithoutResponse(null, channel, registerResourceManagerRequestMessage);
		} catch (FrameworkException e)
		{
			if (e.getFrameworkErrorCode() == FrameworkErrorCode.ChannelIsNotWritable && serverAddress != null)
			{
				releaseChannel(channel, serverAddress);
				if (log.isInfoEnabled())
				{
					log.info("remove channel:" + channel);
				}
			} else
			{
				log.error("", "register failed", e);
			}
		} catch (TimeoutException e)
		{
			log.error(e.getMessage());
		}
	}

	//

	@Override
	public void onRegisterMessageFail(String serverAddress, Channel channel, Message requestMessage, Object response)
	{

		if (response instanceof RegisterResourceManagerResponseMessage && log.isInfoEnabled())
		{
			log.info("register resource_manager_client failed. transaction_coordinator version:" + ((RegisterResourceManagerResponseMessage) response).getVersion());
		}
		throw new FrameworkException("register resource_manager_client failed.");
	}

	//
	//
	public String getResourceIds(IResourceManager resourceManager)
	{
		Map<String, Resource> resourceIdToResourceMap = resourceManager.getResourceIdToResourceMap();
		Set<String> resourceIdSet = resourceIdToResourceMap.keySet();
		if (!resourceIdSet.isEmpty())
		{
			StringBuilder sb = new StringBuilder();
			boolean first = true;
			for (String resourceId : resourceIdSet)
			{
				if (first)
				{
					first = false;
				} else
				{
					sb.append(resourceIds_split_char);
				}
				sb.append(resourceId);
			}
			return sb.toString();
		}
		return null;
	}

	//
	public void releaseChannel(Channel channel, String serverAddress)
	{
		if (null == channel || null == serverAddress)
		{
			return;
		}
		Object connectLock = channelToLockMap.get(serverAddress);
		try
		{
			synchronized (connectLock)
			{
				Channel $channel = serverAddressToChannelMap.get(serverAddress);
				if (null == $channel)
				{
					keyToChannelMap.returnObject(serverAddressToKeyMap.get(serverAddress), channel);
					return;
				}
				if ($channel.compareTo(channel) == 0)
				{
					returnChannel(serverAddress, channel);
				} else
				{
					keyToChannelMap.returnObject(serverAddressToKeyMap.get(serverAddress), channel);
				}
			}
		} catch (Exception exx)
		{
			log.error(exx.getMessage());
		}
	}

	@Override
	public void returnChannel(String serverAddress, Channel channel)
	{
		if (null == channel)
		{
			return;
		}
		try
		{
			if (channel.equals(serverAddressToChannelMap.get(serverAddress)))
			{
				serverAddressToChannelMap.remove(serverAddress);
			}
			keyToChannelMap.returnObject(serverAddressToKeyMap.get(serverAddress), channel);
		} catch (Exception e)
		{
			Log.error(log, Thread.currentThread(), "return channel to rmPool error", e);
		}
	}

}
